package com.study.iceberg;

import com.google.common.collect.ImmutableMap;
import org.apache.hadoop.conf.Configuration;
import org.apache.iceberg.*;
import org.apache.iceberg.catalog.Catalog;
import org.apache.iceberg.catalog.TableIdentifier;
//import org.apache.iceberg.flink.TableLoader;
//import org.apache.iceberg.flink.sink.FlinkSink;
//import org.apache.iceberg.hadoop.HadoopCatalog;
import org.apache.iceberg.hadoop.HadoopCatalog;
import org.apache.iceberg.jdbc.JdbcCatalog;
import org.apache.iceberg.types.Types;

import java.util.HashMap;
import java.util.Map;

public class StreamAPIWriteIceberg {

    public static void main(String[] args) throws Exception {
        //  创建Hadoop配置、Catalog配置和表的Schema，方便后续向路径写数据时可以找到对应的表
        Configuration conf = new Configuration();
        conf.set("fs.s3a.aws.credentials.provider", "org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider");
        conf.set("fs.s3a.connection.ssl.enabled", "false");
        conf.set("fs.s3a.endpoint", "http://127.0.0.1:9000");
        conf.set("fs.s3a.access.key", "minioadmin");
        conf.set("fs.s3a.secret.key", "minioadmin");
        conf.set("fs.s3a.path.style.access", "true");
        conf.set("fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem");
        conf.set("fs.s3a.fast.upload", "true");
        final String warehousePath = "s3a://test/";//minio bucket 路径
        Catalog catalog = new HadoopCatalog(conf, warehousePath);

        //配置iceberg 库名和表名
        TableIdentifier name = TableIdentifier.parse("iceberg_tables");
// SELECT * FROM iceberg_tables WHERE catalog_name = ? AND table_namespace = ? AND table_name = ?
        //创建Iceberg表Schema
        Schema schema = new Schema(
                Types.NestedField.required(1, "id", Types.IntegerType.get()),
                Types.NestedField.required(2, "name", Types.StringType.get()),
                Types.NestedField.required(3, "age", Types.IntegerType.get()),
                Types.NestedField.required(4, "loc", Types.StringType.get()));

        //如果有分区指定对应分区，这里“loc”列为分区列，可以指定unpartitioned 方法不设置表分区
//        PartitionSpec spec = PartitionSpec.unpartitioned();
        PartitionSpec spec = PartitionSpec.builderFor(schema).identity("loc").build();

        //指定Iceberg表数据格式化为Parquet存储
        Map<String, String> props = ImmutableMap.of(TableProperties.DEFAULT_FILE_FORMAT, FileFormat.PARQUET.name());
        Table table;
        // 通过catalog判断表是否存在，不存在就创建，存在就加载
        if (catalog.tableExists(name)) {
            table = catalog.loadTable(name);
        } else {
            table = catalog.createTable(name, schema, spec, props);
        }

       // TableLoader tableLoader = TableLoader.fromHadoopTable("hdfs://mycluster/flink_iceberg/icebergdb/flink_iceberg_tbl", hadoopConf);

        // 向Iceberg中写入数据

    }
}
